package org.aksw.xoperator.sparql.p2p;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ThreadPoolExecutor;

import org.aksw.xoperator.roster.IRosterManager;
import org.aksw.xoperator.sparql.AbstractThreadPooledSparqler;
import org.aksw.xoperator.sparql.ISparqlEndpoint;
import org.aksw.xoperator.sparql.SparqlEndpointIdentifier;
import org.aksw.xoperator.sparql.SparqlEndpointType;
import org.aksw.xoperator.sparql.SparqlExecutionFuture;
import org.aksw.xoperator.sparql.SparqlQuery;
import org.aksw.xoperator.sparql.SparqlResult;
import org.aksw.xoperator.sparql.local.LocalSparqlEndpoint;
import org.aksw.xoperator.xmpp.JabberClientManager;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jivesoftware.smack.PacketCollector;
import org.jivesoftware.smack.SmackConfiguration;
import org.jivesoftware.smack.filter.PacketIDFilter;
import org.jivesoftware.smack.packet.Packet;
import org.jivesoftware.smack.packet.XMPPError;

public class P2PFacade extends AbstractThreadPooledSparqler{
	
	JabberClientManager jabberClientManager;
	
	IRosterManager rmanager;
	
	LocalSparqlEndpoint localSparqlEndpoint;
	
	ThreadPoolExecutor executor;
	
	
	public P2PFacade(JabberClientManager jabberClientManager,
			IRosterManager rmanager, LocalSparqlEndpoint localSparqlEndpoint,
			ThreadPoolExecutor executor) {
		super(executor);
		this.jabberClientManager = jabberClientManager;
		this.rmanager = rmanager;
		this.localSparqlEndpoint = localSparqlEndpoint;

		
		

		
	}
	
	public List<SparqlExecutionFuture<SparqlResult>> getFuturesFor(SparqlQuery query){
		List<SparqlExecutionFuture<SparqlResult>> queries = new ArrayList<SparqlExecutionFuture<SparqlResult>>();
		List<String> sendTo = new ArrayList<String>();
		//send the queries to Buddies
		
		for (String jid : rmanager.getAgentJids()) {
			
			if(query.getAgentsToAsk()==null){
				sendTo.add(jid);
			}else{
				for (SparqlEndpointIdentifier agent : query.getAgentsToAsk()) {
					for (String agentJid : rmanager.getAgentJids()) {
						if(agentJid.startsWith(agent.getName())){
							sendTo.add(agentJid);
						}
					}
					
				}
			}
		}
		//wrap into the Future Tasks
		for (String jid : sendTo) {
			SparqlExecutionFuture<SparqlResult> futureQuery = new SparqlExecutionFuture<SparqlResult>(new P2PCallable(query,jid),jid,query.getQueryString());
			queries.add(futureQuery);
		}
		
		return queries;
	}



	public void process(SparqlQuery query) {
		
		List<SparqlExecutionFuture<SparqlResult>> queries = getFuturesFor(query);

		processFutures(queries,SmackConfiguration.getPacketReplyTimeout());

		
	}
	
	
	
	private class P2PCallable implements Callable<SparqlResult>{
		
		SparqlQuery origQuery;
		
		SPARQLQueryOverXmpp query;
		
		String recipientJid;
		
		private Log log = LogFactory.getLog(P2PCallable.class);
		

		
		public P2PCallable(SparqlQuery query, String recipientJid) {
			super();
			this.origQuery = query;
			this.query = new SPARQLQueryOverXmpp(query);
			this.query.setTo(recipientJid);
			this.recipientJid = recipientJid;
			
		}

		public SparqlResult call() throws Exception {
			PacketCollector collect = jabberClientManager.createPacketCollector(new PacketIDFilter(query.getPacketID()));
			log.info("sending query to " + query.getTo());
			SparqlResult sresult = null;
			sresult = new SparqlResult();
			sresult.setAnsweredBy(new SparqlEndpointIdentifier(
					SparqlEndpointType.P2P, query.getTo()));
			origQuery.addResult(sresult);
			jabberClientManager.sendPacket(query);
			
			Packet packet = collect.nextResult(SmackConfiguration.getPacketReplyTimeout());
			
			if (packet!=null) {
				if (packet.getError() != null) {
					XMPPError error = packet.getError();
					String errorMsg = "An error occured, while asking a peer : "
							+ error.getMessage();
					log.error(errorMsg);

					sresult.setErrorMsg(errorMsg);
				} else {
					SPARQLResultoverXmpp result = (SPARQLResultoverXmpp) packet;

					sresult.setAnswer(result.getResult());
					log.info("Got a Response from the peer : "
							+ result.getResult());
				}
			}
			return sresult;
		}
	}


}
